package com.niodata.dp.task.service.impl;

import com.google.common.base.Joiner;
import com.niodata.dp.core.ConfigurationKeys;
import com.niodata.dp.core.DpConfiguration;
import com.niodata.dp.core.exception.InternalException;
import com.niodata.dp.core.exception.InvalidParamException;
import com.niodata.dp.core.resource.ResourceMapper;
import com.niodata.dp.core.security.ResourceAuthMapper;
import com.niodata.dp.core.usermgr.op.GroupInfoService;
import com.niodata.dp.core.usermgr.op.UserService;
import com.niodata.dp.log.LogUtil;
import com.niodata.dp.task.PluginDocumentGenerator;
import com.niodata.dp.task.PluginPackageParserV2;
import com.niodata.dp.task.dao.JobMapper;
import com.niodata.dp.task.dao.PluginDocumentMapper;
import com.niodata.dp.task.dao.PluginPackageMapperV2;
import com.niodata.dp.task.entity.PluginCategoryCountItem;
import com.niodata.dp.task.entity.PluginCategoryCountSta;
import com.niodata.dp.task.entity.PluginDocument;
import com.niodata.dp.task.entity.PluginDocumentView;
import com.niodata.dp.task.entity.PluginSysCategory;
import com.niodata.dp.task.entity.v2.PluginPackageV2;
import com.niodata.dp.task.entity.v2.PluginPackageViewV2;
import com.niodata.dp.task.service.PluginManagerV2;
import com.niodata.dp.util.FileUtil;
import com.niodata.dt.fs.OssFileSystemBuilder;
import com.niodata.dt.fs.S3FileSystemBuilder;
import com.niodata.dt.fs.Us3FileSystemBuilder;
import freemarker.template.Template;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import javax.annotation.PostConstruct;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;


@Service("PluginManagerV2Impl")
@Transactional
public class PluginManagerV2Impl implements PluginManagerV2 {

  private static Logger logger = Logger.getLogger(PluginManagerV2Impl.class);

  @Autowired
  @Qualifier("DpUserMgrUser")
  private UserService userOperation;
  @Value("${" + ConfigurationKeys.TASK_PACKAGE_DIR + ":/usr/local/dp/package.dir}")
  private String packageDir;

  @Autowired
  @Qualifier("DpUserMgrGroup")
  private GroupInfoService groupOperation;

  @Autowired
  private ResourceMapper resourceMapper;

  @Autowired
  private ResourceAuthMapper resourceAuthMapper;

  @Autowired
  private JobMapper jobMapper;

  @Autowired
  private PluginPackageMapperV2 pluginMapper2;


  @Autowired
  private PluginDocumentMapper documentMapper;

  private FileSystem fileSystem = null;

  private UserGroupInformation ugi;

  private String localStorePath;

  private String remoteStorePath;

  private Template pluginMdTemplate = null;

  /**
   * init .
   */
  @PostConstruct
  public void init() {

    freemarker.template.Configuration cfg = new freemarker.template.Configuration(
        freemarker.template.Configuration.VERSION_2_3_23);
    cfg.setClassForTemplateLoading(PluginManagerV2Impl.class, "/");
    cfg.setDefaultEncoding("UTF-8");

    try {
      pluginMdTemplate = cfg.getTemplate("plugin_md_doc_v2.ftlh");
    } catch (IOException e) {
      LogUtil.logStackTrace(logger, e);
    }

    if (packageDir == null) {
      packageDir = System.getProperty("package.dir");
    }
    if (packageDir == null) {
      packageDir = System.getProperty("user.dir") + File.separator + "package.dir";
    }
    if (packageDir.startsWith("hdfs") || packageDir.startsWith("oss") || packageDir
        .startsWith("s3") || packageDir.startsWith("us3")) {
      remoteStorePath = packageDir;
      //hdfs://ns/
      //oss://endpoint/bucket
      //s3://endpoint/bucket
      String s = packageDir.split("//")[1];
      localStorePath = s.substring(s.indexOf("/"));
    } else if (packageDir.startsWith("file")) {
      String s = packageDir.split("//")[1];
      localStorePath = s;
      remoteStorePath = s;
    } else {
      localStorePath = packageDir;
      remoteStorePath = packageDir;
    }
    try {
      this.fileSystem = createFileSystem(packageDir);
      Path path = new Path(packageDir);
      if (fileSystem.exists(path) && fileSystem.isFile(path)) {
        throw new RuntimeException(packageDir + " is a file");
      }
      if (!fileSystem.exists(path)) {
        fileSystem.mkdirs(path);
      }
    } catch (Exception e) {
      logger.error("create fileSystem error", e);
      throw new RuntimeException("create fileSystem error", e);
    }
  }

  /**
   * PluginManager2Impl.
   */
  public PluginManagerV2Impl() {

  }


  private FileSystem createFileSystem(String packageDir) throws Exception {
    if (packageDir.startsWith("/") || packageDir.startsWith("file://")) {
      FileSystem fs = new LocalFileSystem();
      URI uri = null;
      if (packageDir.startsWith("/")) {
        uri = new URI("file://" + packageDir);
      } else if (packageDir.startsWith("file://")) {
        uri = new URI(packageDir);
      }
      fs.initialize(uri, new Configuration());
      return fs;
    } else if (packageDir.startsWith("hdfs://")) {
      System.setProperty("hadoop.home.dir",
          DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME));
      String userName =
          DpConfiguration.getString(ConfigurationKeys.HDFS_PACKAGE_USER, null);
      if (userName == null) {
        userName = DpConfiguration.getString(ConfigurationKeys.HDFS_PROXY_USER, "hadoop");
      }
      if (DpConfiguration.isHadoopKerberosEnabled()) {
        System.setProperty("java.security.krb5.conf",
            DpConfiguration.getString(ConfigurationKeys.KERBEROS_CONF));
        String keytabPath = DpConfiguration
            .getString(ConfigurationKeys.HDFS_PACKAGE_USER_KEYTAB, null);
        if (keytabPath == null) {
          keytabPath = DpConfiguration.getString(ConfigurationKeys.HDFS_PROXY_USER_KEYTAB);
        }

        Configuration ugiConf = new Configuration();
        ugiConf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION,
            AuthenticationMethod.KERBEROS.name());
        UserGroupInformation.setConfiguration(ugiConf);
        ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(userName, keytabPath);
      } else {
        ugi = UserGroupInformation.createRemoteUser(userName);
      }
      try {
        ugi.doAs(
            new PrivilegedExceptionAction<Void>() {

              @Override
              public Void run() throws Exception {
                Configuration conf = new Configuration();
                conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
                String coreSite =
                    DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME)
                        + "/etc/hadoop/core-site.xml";
                String hdfsSite =
                    DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME)
                        + "/etc/hadoop/hdfs-site.xml";
                conf.addResource(new Path(coreSite));
                conf.addResource(new Path(hdfsSite));
                URI uri;
                try {
                  uri = new URI(DpConfiguration.getString(ConfigurationKeys.HDFS_URI));
                } catch (URISyntaxException ex) {
                  throw new RuntimeException(ex);
                }
                fileSystem = FileSystem.get(uri, conf);
                return null;
              }
            });
      } catch (InterruptedException ex) {
        throw new IOException(ex);
      }
      return fileSystem;
    } else if (packageDir.startsWith("oss://")) {
      OssFileSystemBuilder builder = new OssFileSystemBuilder();
      Map<String, String> conf =
          DpConfiguration.getConfByPrefix(PKG_STORE_CONF_PREFIX);
      return builder.createFileSystem("nouser", conf);
    } else if (packageDir.startsWith("s3://")) {
      S3FileSystemBuilder builder = new S3FileSystemBuilder();
      Map<String, String> conf =
           DpConfiguration.getConfByPrefix(PKG_STORE_CONF_PREFIX);
      return builder.createFileSystem("nouser", conf);
    } else if (packageDir.startsWith("us3://")) {
      Us3FileSystemBuilder builder = new Us3FileSystemBuilder();
      Map<String, String> conf =
          DpConfiguration.getConfByPrefix(PKG_STORE_CONF_PREFIX);
      return builder.createFileSystem("nouser", conf);
    }
    return null;
  }

  @Override
  public PluginCategoryCountSta getPluginCategoryCount(long userId) {
    List<PluginCategoryCountItem> list = this.pluginMapper2.getPluginCategoryCount(userId);
    PluginCategoryCountSta sta = new PluginCategoryCountSta(list);
    return sta;
  }

  @Override
  public PluginCategoryCountSta getPublicToHubPluginCategoryCount() {
    List<PluginCategoryCountItem> list = this.pluginMapper2.getPublicToHubPluginCategoryCount();
    PluginCategoryCountSta sta = new PluginCategoryCountSta(list);
    return sta;
  }

  @Override
  public List<PluginPackageViewV2> getPlugins(long userId, String sysCategory,
      String funcCategory, String nameContains, int offset, int maxCount) {
    return pluginMapper2.getUserAuthorizedPlugins(userId,
        sysCategory, funcCategory, nameContains, offset, maxCount);
  }

  @Override
  public List<PluginPackageViewV2> getPluginsPlublicToHub(String sysCategory,
      String funcCategory, String nameContains, int offset, int maxCount) {
    return pluginMapper2.getPluginsPublicToHub(
        sysCategory, funcCategory, nameContains, offset, maxCount);
  }

  @Override
  public PluginPackageV2 getPluginByNameVersion(long userid, String name, String version) {
    return pluginMapper2.getPluginByNameVersion(name, version);
  }

  @Override
  public PluginDocument getPluginDocument(long userid, String name, String version) {
    return documentMapper.getDocument(name, version);
  }

  @Override
  public PluginDocumentView getPluginDocumentView(long userId, String name, String version) {
    List<PluginPackageV2> histList =
        this.pluginMapper2.getPluginByName(name).stream().filter(new Predicate<PluginPackageV2>() {
          @Override
          public boolean test(PluginPackageV2 pluginPackage) {
            return pluginPackage.getPkgVersion().compareTo(version) <= 0;
          }
        }).collect(Collectors.toList());
    Collections.sort(histList, new Comparator<PluginPackageV2>() {
      @Override
      public int compare(PluginPackageV2 o1, PluginPackageV2 o2) {
        return o1.getPkgVersion().compareTo(o2.getPkgVersion());
      }
    });

    PluginDocumentView view = new PluginDocumentView();

    PluginPackageV2 pkg = this.pluginMapper2.getPluginByNameVersion(name, version);
    view.setDependencies(pkg.getDependencies() != null
        ? pkg.getDependencies().getList()
        : Collections.emptyList());
    view.setLastUpdateTime(pkg.getUpdateTime());
    view.setMaintainers(pkg.getMaintainers());
    view.setPkgId(pkg.getPkgId());
    view.setPkgName(name);
    view.setVersion(version);
    view.setScmUrl(pkg.getSourceRepo());
    view.setPkgLang(pkg.getPkgLang());
    view.setTaskCount(pkg.getTaskCount());
    view.setMinPlatformVersion(pkg.getMinPlatformVersion());
    view.setScmType(pkg.getScmType());
    view.setStatus(pkg.getStatus().name());

    PluginDocument doc = this.documentMapper.getDocument(name, version);
    view.setUserDoc(doc.getDocument());
    return view;
  }

  @Override
  public PluginDocumentView getPluginDocumentView(long userId, String pkgId) {
    PluginPackageV2 pkg = this.pluginMapper2.getPluginByPkgId(pkgId);
    return this.getPluginDocumentView(userId, pkg.getPkgName(), pkg.getPkgVersion());
  }

  @Override
  public void updateDocument(PluginDocument document) {
    documentMapper.updateDocument(document);
  }

  @Override
  public void savePlugin(PluginPackageV2 jobPackage, String document) {
    PluginPackageV2 previousCurrent = this.pluginMapper2
        .getPluginCurrentVersion(jobPackage.getPkgName());
    jobPackage.setCurrent(true);
    pluginMapper2.addPlugin(jobPackage);
    if (previousCurrent != null) {
      previousCurrent.setCurrent(false);
      pluginMapper2.update(previousCurrent);
    }
    PluginDocument pluginDocument = new PluginDocument();
    pluginDocument.setName(jobPackage.getPkgName());
    pluginDocument.setVersion(jobPackage.getPkgVersion());
    pluginDocument.setPkgId(jobPackage.getPkgId());
    pluginDocument.setDocument(document);
    documentMapper.saveDocument(pluginDocument);
  }

  @Override
  public List<PluginPackageV2> getAllPlugins() {
    return pluginMapper2.getAllPlugins();
  }

  @Override
  public PluginPackageV2 getPlugin(String id) {
    return pluginMapper2.getPluginByPkgId(id);
  }

  @Override
  public PluginPackageV2 getPlugin(String name, String version) {
    System.out.println("manager v2");
    return pluginMapper2.getPluginByNameVersion(name, version);
  }

  @Override
  public boolean isRemoteStoreDfs() {
    return !(packageDir.startsWith("/") || packageDir.startsWith("file"));
  }

  @Override
  public void removePlugin(String name, String version) throws IOException {
    PluginPackageV2 pluginPackage = this.pluginMapper2.getPluginByNameVersion(name, version);
    if (pluginPackage != null) {
      if (jobMapper.existsPlugin(pluginPackage.getPkgId())) {
        throw new InternalException("plugin being in used");
      }
      this.pluginMapper2.deletePluginByPkgId(pluginPackage.getPkgId());
      this.documentMapper.deleteDocument(pluginPackage.getPkgName(), pluginPackage.getPkgVersion());
      List<PluginPackageV2> packages = pluginMapper2.getPluginByName(name);
      if (packages.size() > 0) {
        PluginPackageV2 latest = packages.get(0);
        latest.setCurrent(true);
        pluginMapper2.update(latest);
      }
      //TO DO modify plugin auth policy
      //this.resourceAuthMapper.deleteResAuth(pkgId, 0, null, null);
      //this.resourceMapper.deleteResource(pkgId);
      this.removePluginFile(name, version);
    }
  }

  @Override
  public List<PluginPackageV2> getPluginsByName(String name) {
    return this.pluginMapper2.getPluginByName(name);
  }

  @Override
  public List<String> getPluginTypes() {
    return pluginMapper2.getPluginTypes();
  }

  @Override
  public List<String> getSysCategories() {
    return pluginMapper2.getSysCategories();
  }

  @Override
  public List<String> getFuncCategories(String sysCategory) {
    if (PluginSysCategory.App.name().equals(sysCategory)) {
      return this.pluginMapper2.getFuncCategories(sysCategory);
    }
    String[] categories = PluginSysCategory.valueOf(sysCategory).getFuncCategories();
    return Arrays.asList(categories);
  }

  @Override
  public void update(PluginPackageV2 pluginPackage) {
    pluginPackage.setUpdateTime(new Date());
    pluginMapper2.update(pluginPackage);
  }

  @Override
  public void updatePluginUseCount(String pkgId) {
    pluginMapper2.updatePluginUseCount(pkgId);
  }

  @Override
  public PluginPackageV2 savePluginFile(InputStream packageStream, String name, String version)
      throws Exception {
    //save to local _tmp dir
    //check plugin file
    //copy to remote
    String storeTmpDir = Joiner.on(File.separator).join(this.localStorePath, "_tmp",
        name + "-" + version);
    File dir = new File(storeTmpDir);
    if (dir.exists()) {
      FileUtil.deleteFileOrDir(dir);
    }
    dir.mkdirs();
    File file = new File(storeTmpDir + File.separator + name + "-" + version + ".zip");
    if (file.exists()) {
      file.delete();
    }

    FileOutputStream fos = new FileOutputStream(file);
    IOUtils.copy(packageStream, fos);
    fos.close();
    packageStream.close();
    PluginPackageV2 packages = PluginPackageParserV2.parse(file, storeTmpDir);
    if (!packages.getPkgVersion().equals(version) || !packages.getPkgName().equals(name)) {
      throw new InvalidParamException("插件名称或版本不正确:"
          + packages.getPkgName() + ":" + packages.getPkgVersion());
    }
    file.delete();
    if (packages.getPkgName().equals(name) && packages.getPkgVersion().equals(version)) {
      removePluginFile(name, version);
      checkUgi();
      File[] files = new File(packages.getPkgPath()).listFiles();
      Path[] paths = new Path[files.length];
      for (int i = 0; i < files.length; i++) {
        paths[i] = new Path(files[i].getAbsolutePath());
      }
      if (null == fileSystem) {
        throw new RuntimeException("fileSystem is null...");
      }
      Path destPath = new Path(Joiner.on(File.separator).join(packageDir,
          name, version, name + "-" + version));
      fileSystem.mkdirs(destPath);

      fileSystem.copyFromLocalFile(true,
          true,
          paths,
          destPath);
    }
    String pkgPath = "";
    if (packageDir.startsWith("hdfs") || packageDir.startsWith("oss") || packageDir
        .startsWith("s3") || packageDir.startsWith("us3")) {
      pkgPath = remoteStorePath + File.separator + name + File.separator + version
          + File.separator + name + "-" + version;
    } else {
      pkgPath = packageDir + File.separator + name + File.separator + version
          + File.separator + name + "-" + version;
    }
    packages.setPkgPath(pkgPath);
    FileUtil.deleteFileOrDir(dir);
    return packages;
  }

  @Override
  public void copyPluginFilesToLocal(String name, String version) throws IOException {
    if (localStorePath.equals(remoteStorePath)) {
      return;
    }
    this.copyPluginFilesToLocal(name, version, localStorePath);
  }

  @Override
  public void copyPluginFilesToLocal(String name, String version, String localStorePath)
      throws IOException {
    checkUgi();
    File localDir = new File(Joiner.on(File.separator).join(localStorePath, name, version));
    if (!localDir.exists()) {
      boolean succ = localDir.mkdirs();
      if (!succ) {
        throw new RuntimeException("create dir failed:" + localDir.getAbsolutePath());
      }
    }
    this.fileSystem.copyToLocalFile(false, new Path(Joiner.on(File.separator)
            .join(remoteStorePath, name, version, name + "-" + version)),
        new Path(localDir.getAbsolutePath()), true);
  }

  @Override
  public String getLocalStorePath() {
    return localStorePath;
  }

  @Override
  public String getPluginLocalStorePath(String name, String version) {
    return Joiner.on("/").join(localStorePath, name, version, name + "-" + version);
  }

  @Override
  public String getRemoteStorePath() {
    return remoteStorePath;
  }

  @Override
  public boolean localExists(String name, String version) {
    String path = Joiner.on(File.separator).join(
        new String[]{
            localStorePath,
            name,
            version,
            name + "-" + version
        }
    );
    return new File(path).exists();
  }

  @Override
  public boolean exists(String name, String version) throws IOException {
    String path = Joiner.on(File.separator).join(
        new String[]{
            remoteStorePath,
            name,
            version,
            name + "-" + version
        }
    );
    checkUgi();
    return fileSystem.exists(new Path(path));
  }

  @Override
  public void removePluginFile(String name, String version) throws IOException {
    //FileUtil.deleteFileOrDir(new File(Joiner.on(File.separator).join(packageDir, name, version)));
    Path path = new Path(Joiner.on(File.separator).join(packageDir, name, version));
    checkUgi();
    if (fileSystem.exists(path)) {
      fileSystem.delete(path, true);
    }
    String localPath = localStorePath + File.separator + name + File.separator + version;
    File local = new File(localPath);
    FileUtil.deleteFileOrDir(local);
  }

  @Override
  public String getPluginStorePath(String name, String version) {
    return Joiner.on(File.separator).join(packageDir, name, version, name + "-" + version);
  }


  @Override
  public void onPluginUseAgeChanged(List<String> pkgIds) {
    for (String pkgId : pkgIds) {
      this.updatePluginUseCount(pkgId);
    }
  }

  private synchronized void checkUgi() throws IOException {
    if (ugi != null) {
      ugi.checkTGTAndReloginFromKeytab();
    }
  }

  /**
   * gen markdown doc.
   */
  @Override
  public String genUsageMd(PluginPackageV2 pluginPackage)
      throws Exception {
    String pathPrefix = pluginPackage.getPkgPath();
    String usagePath = pathPrefix + File.separator + "doc" + File.separator + "usage.md";
    Path usageFile = new Path(usagePath);
    checkUgi();
    if (!fileSystem.exists(usageFile)) {
      logger.error("pathPrefix:" + pathPrefix);
      logger.error("usage.md path:" + usageFile.toString());
      throw new InvalidParamException("usage.md not exists!");
    }

    FSDataInputStream in = fileSystem.open(usageFile);
    String usageStr = IOUtils.toString(in, "UTF-8");
    return PluginDocumentGenerator.genUsageMd(pluginPackage, usageStr);
  }
}
